package com.mashibing.workqueues;
import com.mashibing.util.RabbitMQConnectionUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消费者类，包含两个测试方法来从RabbitMQ队列接收消息
 */
public class Consumer1 {
    /**
     * 消费者1接收消息的测试方法
     * 该方法建立与RabbitMQ的连接，监听指定队列，并在接收到消息后进行处理
     * @throws IOException 如果建立连接或读取消息时发生I/O错误
     * @throws TimeoutException 如果建立连接时发生超时
     * @throws InterruptedException 如果线程睡眠被中断
     */
    @Test
    public void consume1() throws IOException, TimeoutException, InterruptedException {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.构建channel
        Channel channel1 = connection.createChannel();

        //3.构建队列
        channel1.queueDeclare(Publisher.QUEUE_NAME, false, false, false, null);
        // 设置通道的预取计数，以控制消息的并发处理数量
        channel1.basicQos(1);

        //4.监听消息
        DefaultConsumer callback = new DefaultConsumer(channel1) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    // 模拟消息处理时间
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // 打印接收到的消息内容
                System.out.println("消费者1收到消息：" + new String(body, "UTF-8"));
                // 确认消息已成功消费
                channel1.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 开始消费消息，不自动确认
        channel1.basicConsume(Publisher.QUEUE_NAME, false , callback);
        // 打印监听开始信息
        System.out.println("开始监听队列");
        // 阻塞当前线程，保持程序运行
        System.in.read();

    }

    /**
     * 消费者2接收消息的测试方法
     * 该方法与消费者1类似，但模拟的消息处理时间更长
     * @throws IOException 如果建立连接或读取消息时发生I/O错误
     * @throws TimeoutException 如果建立连接时发生超时
     * @throws InterruptedException 如果线程睡眠被中断
     */
    @Test
    public void consume2() throws IOException, TimeoutException, InterruptedException {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.构建channel
        Channel channel2 = connection.createChannel();

        //3.构建队列
        channel2.queueDeclare(Publisher.QUEUE_NAME, false, false, false, null);
        // 设置通道的预取计数，以控制消息的并发处理数量
        channel2.basicQos(1);

        //4.监听消息
        DefaultConsumer callback = new DefaultConsumer(channel2) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    // 模拟消息处理时间
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // 打印接收到的消息内容
                System.out.println("消费者2收到消息：" + new String(body, "UTF-8"));
                // 确认消息已成功消费
                channel2.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 开始消费消息，不自动确认
        channel2.basicConsume(Publisher.QUEUE_NAME, false, callback);
        // 打印监听开始信息
        System.out.println("开始监听队列");
        // 阻塞当前线程，保持程序运行
        System.in.read();

    }
}

